package com.atguigu.bigdata.gmall.realtime.util;


import com.atguigu.bigdata.gmall.realtime.common.Constant;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
import org.apache.kafka.clients.consumer.ConsumerRecord;

import java.nio.charset.StandardCharsets;
import java.util.Properties;

/**
 * @Author lzc
 * @Date 2022/10/5 16:15
 */
public class FlinkSourceUtil {
    public static SourceFunction<String> getKafkaSource(String topic, String groupId) {
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", Constant.KAFKA_BROKERS);
        props.setProperty("group.id", groupId);
        FlinkKafkaConsumer<String> source = new FlinkKafkaConsumer<>(
            topic,
            new KafkaDeserializationSchema<String>() {
                // 是否要结束流
                @Override
                public boolean isEndOfStream(String nextElement) {
                    // 从 kafka 消费数据, 流永远不停, 返回 false
                    return false;
                }
                
                // 对传入的数组, 实现反序列化
                @Override
                public String deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
                    byte[] value = record.value();
                    // 如果 value 是 null, 不需要反序列化
                    if (value != null) {
                        return new String(value, StandardCharsets.UTF_8);
                    }
                    return null;
                }
                
                // 读到数据的类型, 决定了流中的数据类型
                @Override
                public TypeInformation<String> getProducedType() {
//                    return Types.STRING;
//                    return TypeInformation.of(String.class);
                    return TypeInformation.of(new TypeHint<String>() {});
                }
            },
            props
        );
        source.setStartFromLatest();
        return source;
    }
}
